package org.apache.rocketmq.example.quickstart;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

/**
 * 普通消息 - 异步发送
 * 通过监听获取发送消息的响应，不会阻塞发送消息的线程
 * 适用于对处理速度要求较高的场景，让生产者可以连续发送消息，而不用阻塞
 */
public class AsyncProducer {
    public static void main(String[] args) {
        // 创建生产者
        DefaultMQProducer producer = new DefaultMQProducer("message-group-2");
        // 设置nameServer地址
        producer.setNamesrvAddr("39.107.66.31:9876");
        // 启动生产者
        try {
            producer.start();
            // 发送消息
            for (int i = 0; i < 10; i++) {
                // 创建消息实体
                Message message = new Message("message-topic-3", "tageA", ("Hello RocketMQ" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                // 设置延时等级 一共有18个等级,等级越大延迟越久
                message.setDelayTimeLevel(1);
                // 发送消息并接收响应
                producer.send(message, new SendCallback() {
                    // 异步监听，通过回调处理发送消息响应
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        System.out.println("发送成功!");
                        System.out.println(sendResult);
                    }

                    @Override
                    public void onException(Throwable e) {
                        System.out.println(e.getMessage());
                    }
                });
            }
            Thread.sleep(10000);
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            // 停止生产者
            producer.shutdown();
        }
    }
}
